package org.example.api.tableapi;

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/3/9
 */
public class TimeAndWindowTest {

    public static void main(String[] args) throws Exception {
        // 1 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        // 2 数据流
        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(SensorReading sensorReading) {
                return sensorReading.getTimestamp() * 1000L;
            }
        });

        // 3 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 4 基于数据流创建表
        // 时间特性设置
//        Table table = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, pt.proctime");
//        Table table = tableEnv.fromDataStream(dataStream, "id, timestamp.rowtime as ts, temperature as temp, rt.rowtime");
        Table table = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, rt.rowtime");
        tableEnv.createTemporaryView("sensor", table);

        // 5 窗口操作
        // 5.1 Group Window
        // table api
        Table resultTable = table.window(Tumble.over("10.seconds").on("rt").as("tw"))
                .groupBy("id, tw")
                .select("id, id.count, temp.avg, tw.end");

        // sql
        Table sqlTable = tableEnv.sqlQuery(
                "select id, count(id) as cnt, avg(temp) as avgTemp, tumble_end(rt, interval '10' second) " +
                        "from sensor group by id, tumble(rt, interval '10' second)"
        );

        table.printSchema();

        // 数据只增不改可用 toAppendStream
//        tableEnv.toAppendStream(resultTable, Row.class).print("result");
//        tableEnv.toRetractStream(sqlTable, Row.class).print("sql");

        // 5.2 Over Window
        // table api
        Table overResultTable = table.window(
                Over.partitionBy("id")
                        .orderBy("rt")
                        .preceding("2.rows")
                        .as("ow")
        ).select("id, rt, id.count over ow, temp.avg over ow");

        // sql
        Table overSqlTable = tableEnv.sqlQuery("select id, rt, count(id) over ow, avg(temp) over ow " +
                "from sensor " +
                "window ow as (partition by id order by rt rows between 2 preceding and current row )");

        tableEnv.toAppendStream(overResultTable, Row.class).print("result");
//        tableEnv.toRetractStream(overSqlTable, Row.class).print("sql");


        env.execute();

    }
}
